Solution: Messaging
Although in the video I’m using Pydantic V1, we’ve updated the code example to Pydantic V2. As a result, there might be minor differences between the code you see in the video and the code in the Git repository. Notably, .dict() has been replaced by .model_dump().
Hi Arjan,
nice solution creating the observer-pattern as a mixture of object oriented and functional programming.
I have choosen the fully functional approach, since I am already familiar with the classic object version in a current project.
from collections import defaultdict
from typing import Callable, Any
from enum import StrEnum, auto
class EventTypes(StrEnum):
EVENT_CREATED = auto()
TICKET_BOOKED = auto()
SUBSCRIBERS: dict[EventTypes, list[Callable[[Any],None]]] = defaultdict(list)
def subscribe(event_type: EventTypes, fn: Callable[[Any], None]):
SUBSCRIBERS[event_type].append(fn)
def post_event(event_type: EventTypes, data: Any):
if event_type not in SUBSCRIBERS:
return
for fn in SUBSCRIBERS[event_type]:
fn(data)
and e.g. the sms_listner:
from typing import Protocol
from event import subscribe, EventTypes
class Ticket(Protocol):
event_id: int
customer_name: str
customer_email: str
def send_SMS(data: str):
print(f"SMS: {data}")
def handle_booked_ticket(ticket: Ticket):
send_SMS(f"Your ticket with id {ticket.event_id} has been booked")
def setup_sms_event_handlers():
subscribe(EventTypes.TICKET_BOOKED, handle_booked_ticket)
I still think about how to solve the typing issue (MessageType already indicates that e.g. TICKET_BOOKED should lead in data being a Ticket) to avoid "# type: ignore" in your example or "Any" in mine.
Nice solution Phillipp!
Regarding the typing issue, this is something that we are looking into as well, hopefully, this is something that can be fixed in the foreseeable future.
Hi Arjan,
is there a reason we need a MessageSystem class that has awareness of all handlers for all MessageTypes, even though create_event and book_ticket each only care about a single one of those message types?
i would be tempted to use a "flatter" class (similar to MessageSystem, but only managing a single list of handlers) that would have one instance defined for each MessageType and would be the only thing you'd need to pass as a dependency to create_event and book_ticket...
pros? cons?
Hi Sylvain. I think both approaches work. I chose this route because having a separate instance for each message type simply means that you need that dictionary that maps message types to handlers somewhere else in the code. And without it, the MessageSystem is nothing more than a list of hander functions, so there's almost no need for a class. So to me, it made more sense to simply put all of the logic into the MessageSystem class.
thanks Arjan! I have question about your MessageSystem class.
What would be the use-case of the detach function? For other administrative users of this api to be able to add and remove message handlers?
I think that I will try to implement this messaging design in a project.
My solution was the plug-in pattern with a dictionary and handler functions added to the operations as well.
Hi David, being able to detach message handlers is quite useful in a live system. For example, let's say you have a handler that sends an email to an admin account whenever someone has a login problem. Once you see that such a problem occurs, you might want to switch off the automatic emails/messages for the time being until you fix the bug, to avoid overrunning your mailbox.